package task

import (
	"encoding/json"
	"errors"
	"io"
	"log"
	"math/rand"
	"net/http"
	"path"
	"scheduling/config"
	"scheduling/model"
	"scheduling/pool"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/bwmarrin/snowflake"
	"github.com/gin-gonic/gin"
	_ "github.com/go-sql-driver/mysql"

	"github.com/jmoiron/sqlx"
)

var (
	defaultLetters = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789")

	db      *sqlx.DB
	genNode *snowflake.Node
	DBS     = make(map[string]*sqlx.DB, 0)
)

func init() {
	c := config.GetConf()
	var err error

	db, err = pool.NewDBConnect(c.Driver.Name, c.Driver.Url, 2, 5)
	if err != nil {
		log.Fatal(err)
	}
	//初始化雪花算法
	genNode, err = snowflake.NewNode(1)
	if err != nil {
		panic(err)
	}
	initDatasources()
}

func initDatasources() {
	data := make([]model.CfDatasources, 0)
	err := db.Select(&data, `select * from cf_datasources`)
	if err != nil {
		panic(err)
	}
	for _, ds := range data {
		curDb, err := pool.NewDBConnect(ds.DriverName.String, ds.SourceUrl.String, ds.MaxIdleConns, ds.MaxOpenConns)
		if err != nil {
			log.Println("初始化数据连接池失败：", ds.DriverName, ds.SourceUrl)
			continue
		}
		DBS[ds.Pid.String] = curDb
	}
}

type FlowData struct {
	NodeList []Node `json:"nodeList"`
	LineList []Line `json:"lineList"`
}

//GetNode 通过节点编号获取节点信息
func GetNode(nodes []Node, nodeId string) *Node {
	for _, n := range nodes {
		if n.Id == nodeId {
			return &n
		}
	}
	return nil
}

//GetLine 通过节点编号获取节点线
func GetLine(lines []Line, nodeId string) []Line {
	ls := make([]Line, 0)
	for _, l := range lines {
		if l.From == nodeId {
			ls = append(ls, l)
		}
	}
	return ls
}

type Line struct {
	From  string `json:"from"`
	To    string `json:"to"`
	Label string `json:"label"`
}

type Param struct {
	Name     string `json:"name"`
	KeyName  string `json:"keyName"`
	KeyValue string `json:"keyValue"`
	SourceId string `json:"sourceId"`
}

type Node struct {
	Id       string            `json:"id"`
	Name     string            `json:"name"`
	NodeType string            `json:"type"`
	Left     string            `json:"left"`
	Top      string            `json:"top"`
	Ico      string            `json:"ico"`
	State    string            `json:"state"`
	Params   []Param           `json:"params"`
	DataType string            `json:"dataType"`
	SavePath string            `json:"savePath"`
	Content  map[string]string `json:"content"`
}

type FlowTask struct {
	wg      sync.WaitGroup
	c       *gin.Context
	out     map[string]map[string]interface{}
	outList map[string][]map[string]interface{}
	outText map[string]string
	outByte map[string][]byte
	Data    *FlowData
	errMsg  []string
}

func NewFlowTask(c *gin.Context, data string) (*FlowTask, error) {
	t := new(FlowTask)
	t.wg = sync.WaitGroup{}
	t.out = make(map[string]map[string]interface{}, 0)
	t.outList = make(map[string][]map[string]interface{}, 0)
	t.outText = make(map[string]string, 0)
	t.outByte = make(map[string][]byte, 0)
	t.c = c
	t.Data = new(FlowData)
	err := json.Unmarshal([]byte(data), t.Data)
	if err != nil {
		log.Println("流程配置不正确！", err)
		return nil, err
	}
	return t, nil
}

//Start　执行开始节点任务
func (m *FlowTask) Start(perms map[string]model.SignKeys) {
	node := m.Data.NodeList[0]
	result := m.BindInput(node.Params)
	data := make(map[string]interface{}, 0)
	key := node.Content["key"]
	permType := node.Content["permType"]
	signType := node.Content["signType"]
	argName := node.Content["argName"]

	switch node.DataType {
	case "json":
		log.Printf("任务JSON数据处理：%s\n", node.Name)
		err := m.c.BindJSON(&data)
		if err != nil {
			log.Println("json参数格式不正确!", err)
			m.c.JSON(http.StatusOK, gin.H{"code": "1002", "msg": "json参数格式不正确!" + err.Error()})
			return
		}
		if key != "" {
			perm := NewPerms(perms[key], permType, signType, argName)
			if !perm.Certification(m.c, data) {
				m.c.JSON(http.StatusOK, gin.H{"code": "1000", "msg": "签名认证失败!"})
				return
			}
		}
	case "form":
		log.Printf("任务FORM数据处理：%s\n", node.Name)

		for _, item := range node.Params {
			data[item.KeyName] = m.c.PostForm(item.KeyName)
		}
		if key != "" {
			perm := NewPerms(perms[key], permType, signType, argName)
			if !perm.Certification(m.c, data) {
				m.c.JSON(http.StatusOK, gin.H{"code": "1000", "msg": "签名认证失败!"})
				return
			}
		}
	case "file":
		log.Printf("任务文件上传处理：%s\n", node.Name)
		for _, item := range node.Params {
			data[item.KeyName] = m.c.PostForm(item.KeyName)
		}
		if key != "" {
			perm := NewPerms(perms[key], permType, signType, argName)
			if !perm.Certification(m.c, data) {
				m.c.JSON(http.StatusOK, gin.H{"code": "1000", "msg": "签名认证失败!"})
				return
			}
		}
		form, err := m.c.MultipartForm()
		if err != nil {
			m.c.JSON(http.StatusOK, gin.H{"code": "1002", "msg": err.Error()})
			return
		}
		files := form.File["file"]
		sourceFile := make([]map[string][]byte, 0)
		for _, file := range files {
			fileExt := path.Ext(file.Filename)
			saveFile := genNode.Generate().String() + fileExt
			f, err := file.Open()
			if err != nil {
				log.Println("文件上传,准备读取文件:", err)
				m.c.JSON(http.StatusOK, gin.H{"code": "1002", "msg": "文件上传,准备读取文件:" + err.Error()})
				return
			}
			bs, err := io.ReadAll(f)
			if err != nil {
				log.Println("文件上传,读取文件:", err)
				m.c.JSON(http.StatusOK, gin.H{"code": "1002", "msg": "文件上传,读取文件:" + err.Error()})
				return
			}
			item := make(map[string][]byte, 0)
			item[saveFile] = bs
			sourceFile = append(sourceFile, item)
		}
		result["sourceFile"] = sourceFile
	}
	for _, p := range node.Params {
		val := data[p.KeyName]
		result[p.KeyName] = val
	}
	m.out[node.Id] = result
	err := m.Next(node.Id)
	if err != nil {
		m.errMsg = append(m.errMsg, err.Error())
	}
	for _, n := range m.Data.NodeList {
		if n.NodeType == "END" {
			m.End(&n)
			return
		}
	}
}

//Next　执行下一步节点任务
func (m *FlowTask) Next(nodeId string) error {
	lines := GetLine(m.Data.LineList, nodeId)
	for _, line := range lines {
		if line.Label != "" {
			ok, err := m.Expression(line.Label, nodeId, nil, nil)
			if err != nil || !ok {
				log.Printf("流程线条件判断：%s,结果：%v,%s", line.Label, ok, err)
				continue
			}
		}
		node := GetNode(m.Data.NodeList, line.To)
		if node.NodeType == "END" {
			return nil
		}
		m.wg.Add(1)
		go m.syncNode(node)
	}
	return nil
}

//syncNode 并行执行多子节点任务
func (m *FlowTask) syncNode(node *Node) {
	defer m.wg.Done()
	err := m.RunNodeType(node)
	if err != nil {
		log.Println(err)
		m.errMsg = append(m.errMsg, err.Error())
		return
	}
	err = m.Next(node.Id)
	if err != nil {
		log.Println(err)
		m.errMsg = append(m.errMsg, err.Error())
		return
	}
}

//RunNodeType 按类型执行流程
func (m *FlowTask) RunNodeType(node *Node) error {
	log.Printf("节点处理：%s %s\n", node.Name, node.NodeType)
	switch node.NodeType {
	case "HTTP":
		handler := NewNetHandler(m)
		return handler.RequestHttp(node)
	case "HTTP_UPLOAD":
		handler := NewNetHandler(m)
		return handler.Upload(node)
		return nil
	case "EXEC_SQL":
		handler := NewDbHandler(m)
		return handler.Exec(node)
	case "QUERY_SQL":
		handler := NewDbHandler(m)
		return handler.Query(node)
	case "FILE_SAVE":
		handler := NewFileHandler(m)
		return handler.Save(node)
	case "FILE_READ":
		handler := NewFileHandler(m)
		return handler.Read(node)
	case "FILE_MOVE":
		handler := NewFileHandler(m)
		return handler.Move(node)
	case "FILE_DELETE":
		handler := NewFileHandler(m)
		return handler.Delete(node)
	case "TCP_CLIENT":
		handler := NewTcpClient(m)
		return handler.Handler(node)
	case "WEBSOCKET":
		return NewWebSocketPush(m, node)
	case "CLOUD_NACOS":
		handler := NewCloudHandler(m)
		return handler.Handler(node)
	}
	return nil
}

// End 流程结束－组装参数
func (m *FlowTask) End(node *Node) {
	m.wg.Wait()
	log.Printf("流程结束数据处理：%s", node.Name)
	if len(m.errMsg) != 0 {
		log.Println(m.errMsg)
		m.c.JSON(http.StatusOK, gin.H{"code": "5000", "msg": strings.Join(m.errMsg, ",")})
		return
	}
	successCode := node.Content["successCode"]
	dataKey := node.Content["dataKey"]
	dataType := node.Content["dataType"]

	if len(node.Params) == 0 {
		m.c.JSON(http.StatusOK, gin.H{"code": successCode, "msg": "成功"})
		return
	}
	result := m.BindInput(node.Params)
	switch dataType {
	case "bytes":
		for _, v := range result {
			bs, ok := v.([]byte)
			if !ok {
				m.c.JSON(http.StatusOK, gin.H{"code": "5000", "msg": "响应参数绑定与响应类型不匹配"})
				return
			}
			m.c.Writer.Write(bs)
			m.c.Writer.Flush()
		}
	default:
		if dataKey != "" {
			m.c.JSON(http.StatusOK, gin.H{"code": successCode, "msg": "成功", dataKey: result})
		} else if len(result) == 1 {
			for k, v := range result {
				if dataType == "string" {
					m.c.String(http.StatusOK, "%s", v)
				} else {
					m.c.JSON(http.StatusOK, gin.H{"code": successCode, "msg": "成功", k: v})
				}
				break
			}
		} else {
			m.c.JSON(http.StatusOK, result)
		}
	}
	return
}

//BindSystem　获取系统参数值
func (m *FlowTask) BindSystem(key string) string {
	switch key {
	case "$DATE-14":
		return time.Now().Format("2006-01-02 15:04:05")
	case "$DATE-8":
		return time.Now().Format("2006-01-02")
	case "$DATE14":
		return time.Now().Format("20060102150405")
	case "$DATE8":
		return time.Now().Format("20060102")
	case "$TIME-6":
		return time.Now().Format("15:04:05")
	case "$TIME6":
		return time.Now().Format("150405")
	case "$SYSTEMID":
		return genNode.Generate().String()
	case "$RANDOM_STR":
		b := make([]rune, 32)
		for i := range b {
			b[i] = defaultLetters[rand.Intn(len(defaultLetters))]
		}
		return string(b)

	}
	return key
}

//BindInput 执行流程－组装输入参数
func (m *FlowTask) BindInput(target []Param) map[string]interface{} {
	result := make(map[string]interface{}, 0)
	for _, item := range target {
		var val interface{}
		if strings.HasPrefix(item.KeyValue, `"`) && strings.HasSuffix(item.KeyValue, `"`) {
			val = item.KeyValue[1 : len(item.KeyValue)-1]
		} else if strings.HasPrefix(item.KeyValue, "$") {
			val = m.BindSystem(item.KeyValue)
		} else if item.SourceId != "" {
			values := m.out[item.SourceId]
			if values != nil && item.KeyValue != "" {
				for k, v := range values {
					if item.KeyValue == k {
						val = v
						break
					}
				}
			} else {
				textValue := m.outText[item.SourceId]
				if textValue != "" {
					val = textValue
				}
				if val == nil {
					listValue := m.outList[item.SourceId]
					if listValue != nil {
						val = listValue
					}
				}
				if val == nil {
					byteValue := m.outByte[item.SourceId]
					if byteValue != nil {
						val = byteValue
					}
				}
			}
		} else {
			for _, in := range m.out {
				for k, v := range in {
					if k == item.KeyValue {
						val = v
						break
					}
				}
			}
		}
		result[item.KeyName] = val
	}
	return result
}

//Expression 执行表达式
func (m *FlowTask) Expression(expr string, frontNodeId string, input map[string]interface{}, inputs []map[string]interface{}) (bool, error) {
	exprs := strings.Split(expr, " ")
	key := strings.TrimSpace(exprs[0])
	exp := strings.TrimSpace(exprs[1])
	value := strings.TrimSpace(exprs[2])
	if input == nil && frontNodeId != "" {
		input = m.out[frontNodeId]
	}
	if len(exprs) != 3 {
		return false, errors.New("表达式不正确！")
	}
	switch exp {
	case "==":
		if key == "$size" {
			v, err := strconv.Atoi(value)
			if err != nil {
				return false, errors.New("表达式值格式不正确！")
			}
			if len(inputs) == v {
				return true, nil
			}
		} else if input[key] == nil && value == `""` {
			return true, nil
		} else if input[key].(string) == value {
			return true, nil
		}
	case "!=":
		if key == "$size" {
			v, err := strconv.Atoi(value)
			if err != nil {
				return false, errors.New("表达式值格式不正确！")
			}
			if len(inputs) != v {
				return true, nil
			}
		} else if input[key] == nil && value == `""` {
			return false, nil
		} else if input[key].(string) != value {
			return true, nil
		}
	case ">":
		v, err := strconv.Atoi(value)
		if err != nil {
			return false, errors.New("表达式参数格式不正确！")
		}
		if key == "$size" {
			if len(inputs) > v {
				return true, nil
			}
		} else if input[key] == nil {
			return false, nil
		} else if input[key].(int) > v {
			return true, nil
		}
	case "<":
		v, err := strconv.Atoi(value)
		if err != nil {
			return false, errors.New("表达式参数格式不正确！")
		}
		if key == "$size" {
			if len(inputs) < v {
				return true, nil
			}
		} else if input[key] == nil {
			return false, nil
		} else if input[key].(int) < v {
			return true, nil
		}
	case "<=":
		v, err := strconv.Atoi(value)
		if err != nil {
			return false, errors.New("表达式参数格式不正确！")
		}
		if key == "$size" {
			if len(inputs) <= v {
				return true, nil
			}
		} else if input[key] == nil {
			return false, nil
		} else if input[key].(int) <= v {
			return true, nil
		}
	case ">=":
		v, err := strconv.Atoi(value)
		if err != nil {
			return false, errors.New("表达式参数格式不正确！")
		}
		if key == "$size" {
			if len(inputs) >= v {
				return true, nil
			}
		} else if input[key] == nil {
			return false, nil
		} else if input[key].(int) >= v {
			return true, nil
		}
	}
	return false, errors.New("表达式验证未通过！")
}
